package top.reminis.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;

/**
 * MinIO连接测试工具
 * 
 * @author zwf
 * @create 2025-09-20
 */
@Slf4j
public class MinIOTestUtils {
    
    /**
     * 测试MinIO连接并确保桶存在
     * 
     * @param endpoint MinIO端点
     * @param accessKey 访问密钥
     * @param secretKey 秘密密钥
     * @param bucket 桶名称
     * @return 连接是否成功
     */
    public static boolean testConnection(String endpoint, String accessKey, String secretKey, String bucket) {
        try {
            log.info("测试MinIO连接...");
            log.info("Endpoint: {}", endpoint);
            log.info("Access Key: {}", accessKey);
            log.info("Secret Key: {}", secretKey.length() > 0 ? "***" : "空");
            log.info("Bucket: {}", bucket);
            
            // 首先使用MinIO客户端确保桶存在
            boolean bucketExists = MinIOBucketManager.ensureBucketExists(endpoint, accessKey, secretKey, bucket);
            if (!bucketExists) {
                log.error("MinIO桶创建失败");
                return false;
            }
            
            // 然后使用Hadoop FileSystem测试S3A连接
            Configuration conf = new Configuration();
            conf.set("fs.s3a.endpoint", endpoint);
            conf.set("fs.s3a.access.key", accessKey);
            conf.set("fs.s3a.secret.key", secretKey);
            conf.set("fs.s3a.path.style.access", "true");
            conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
            conf.set("fs.s3a.connection.ssl.enabled", "false");
            conf.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
            
            String s3aPath = "s3a://" + bucket + "/";
            URI uri = URI.create(s3aPath);
            FileSystem fs = FileSystem.get(uri, conf);
            
            // 确保iceberg目录存在（与配置文件中的warehouse路径一致）
            try {
                String icebergPath = s3aPath + "iceberg/";
                boolean icebergExists = fs.exists(new Path(icebergPath));
                if (!icebergExists) {
                    log.info("iceberg目录不存在，尝试创建...");
                    boolean icebergCreated = fs.mkdirs(new Path(icebergPath));
                    if (icebergCreated) {
                        log.info("iceberg目录创建成功");
                    } else {
                        log.warn("iceberg目录创建失败，但继续执行");
                    }
                } else {
                    log.info("iceberg目录已存在");
                }
            } catch (Exception e) {
                log.warn("检查iceberg目录时发生错误，但继续执行: {}", e.getMessage());
                // 尝试直接创建iceberg目录
                try {
                    String icebergPath = s3aPath + "iceberg/";
                    boolean icebergCreated = fs.mkdirs(new Path(icebergPath));
                    if (icebergCreated) {
                        log.info("iceberg目录创建成功");
                    } else {
                        log.warn("iceberg目录创建失败，但继续执行");
                    }
                } catch (Exception e2) {
                    log.warn("创建iceberg目录时发生错误，但继续执行: {}", e2.getMessage());
                }
            }
            
            log.info("MinIO连接测试结果: 成功");
            
            fs.close();
            return true;
            
        } catch (Exception e) {
            log.error("MinIO连接测试失败", e);
            return false;
        }
    }
    
    /**
     * 创建测试目录
     * 
     * @param endpoint MinIO端点
     * @param accessKey 访问密钥
     * @param secretKey 秘密密钥
     * @param bucket 桶名称
     * @param testPath 测试路径
     * @return 创建是否成功
     */
    public static boolean createTestDirectory(String endpoint, String accessKey, String secretKey, 
                                            String bucket, String testPath) {
        try {
            log.info("创建测试目录: {}", testPath);
            
            Configuration conf = new Configuration();
            conf.set("fs.s3a.endpoint", endpoint);
            conf.set("fs.s3a.access.key", accessKey);
            conf.set("fs.s3a.secret.key", secretKey);
            conf.set("fs.s3a.path.style.access", "true");
            conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
            conf.set("fs.s3a.connection.ssl.enabled", "false");
            conf.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
            
            String s3aPath = "s3a://" + bucket + "/" + testPath;
            URI uri = URI.create(s3aPath);
            FileSystem fs = FileSystem.get(uri, conf);
            
            Path path = new Path(s3aPath);
            boolean created = fs.mkdirs(path);
            log.info("测试目录创建结果: {}", created ? "成功" : "失败");
            
            fs.close();
            return created;
            
        } catch (Exception e) {
            log.error("创建测试目录失败", e);
            return false;
        }
    }
}
